package main

//kafka消费者
import (
	"fmt"
	"github.com/Shopify/sarama"
	"sync"
)

func main() {
	consumer, err := sarama.NewConsumer([]string{"192.168.56.129:9092", "192.168.56.130:9092", "192.168.56.131:9092"}, nil)
	if err != nil {
		fmt.Printf("FAILED TO START consumer,err:%v\n", err)
		return
	}
	var wg sync.WaitGroup
	//找到topic所有的分区列表
	paratitionList, err := consumer.Partitions("es_log")
	if err != nil {
		fmt.Printf("failed to get list os paratition:err:%v\n", err)
		return
	}
	fmt.Println(paratitionList)
	//遍历分区，针对每个分区读取消息
	for paration := range paratitionList {
		//针对每个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition("es_log", int32(paration), sarama.OffsetNewest)
		if err != nil {
			fmt.Printf("failed to start consumer for paraition %d,err:%v\n", paration, err)
			return
		}
		defer pc.AsyncClose()
		//异步从每个分区获取消费信息
		wg.Add(1)
		go func(partitioner sarama.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("parition:%d Offset:%d key:%s value:%s", msg.Partition, msg.Offset, msg.Key, msg.Value)
			}
		}(pc)
	}
	wg.Wait()
}
